package com.custom;


import com.custom.source.ClinkReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordEmitter;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;

import java.util.Map;
import java.util.function.Supplier;

/**
 * 具体的源阅读器
 */
public class MyReader<E, T>
        extends ClinkReader<E, T, MySplit, MySplitState> {
    public MyReader(Supplier<SplitReader<E, MySplit>> splitReaderSupplier,
                    RecordEmitter<E, T, MySplitState> recordEmitter,
                    Configuration config,
                    SourceReaderContext context) {
        super(splitReaderSupplier, recordEmitter, config, context);
    }

    /**
     * 处理一些完成的分片,比如清理状态
     */
    @Override
    protected void onSplitFinished(Map<String, MySplitState> finishedSplitIds) {
        context.sendSplitRequest();
    }

    /**
     * 当新的分片被加入到reader,给这个分片初始化状态
     */
    @Override
    public MySplitState initializedState(MySplit split) {
        return new MySplitState(split);

    }

    /**
     * 将可变的分片状态转换为不可变的分片
     */
    @Override
    protected MySplit toSplitType(String splitId, MySplitState splitState) {
        return splitState.getState();
    }

}
